package portrait

import org.apache.commons.lang.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.Row

object KeyTags extends Tags {
  override def makeTags(any: Any*): List[(String, Int)] = {
    val row = any(0).asInstanceOf[Row]
    val stopBro = any(1).asInstanceOf[Broadcast[Map[String,Null]]]
    val keywords = row.getAs[String]("keywords")
    var list = List[(String, Int)]()
    keywords.split("\\|",-1).filter(str=>str.length>=3 && str.length<=8 && !stopBro.value.contains(str))
      .foreach(k=>list :+= ("K"+k,1))


    val pname = row.getAs[String]("provincename")
    val cname = row.getAs[String]("cityname")
    if (StringUtils.isNotEmpty(pname)){
      list :+= ("ZP"+pname,1)
    }
    if (StringUtils.isNotEmpty(cname)){
      list :+= ("ZC"+cname,1)
    }
    list
  }
}
